package com.intct.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.intct.hbase.bean.AreaBean;
import com.intct.hbase.bean.VehiceBean;
import com.intct.func.HBaseSink;
import com.intct.ods.OdsManApp;
import com.intct.util.KafkaUtil;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.HashSet;
import java.util.Set;

/**
 * @author gufg
 * @since 2025-08-25 14:18
 * 车辆分布 (区域)
 */
public class DwsVechicleApp {
    public static void main(String[] args) throws Exception {
        // 从命令中获取参数
        ParameterTool paraTools = ParameterTool.fromArgs(args);
        String mvnProjectPath = OdsManApp.class.getResource("/test.properties").getFile();

        String filePath = paraTools.get("filePath", mvnProjectPath);
        String topic = paraTools.get("kafka-topic-name", "ods_locus");

        // 从配置文件中获取配置参数
        ParameterTool propertiesFile = ParameterTool.fromPropertiesFile(filePath);

        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.enableCheckpointing(6000L, CheckpointingMode.EXACTLY_ONCE);

        // 配置源 并获取数据源数据
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(propertiesFile, "dwd_vehicle", "dws_vechicle_group_id");
        DataStreamSource<String> sourceDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "dws_vechicle_group_id");


        /*
            {
              "actionTime": 1753778146,
              "areaId": "110102",
              "code": "110102",
              "driverId": "934",
              "carShelfNumber": "Y479045282D",
              "car": "926",
              "name": "北京市西城区",
              "type": "0",
              "speed": 50
            }
         */
        SingleOutputStreamOperator<VehiceBean> mapDS = sourceDS.map(new MapFunction<String, VehiceBean>() {
            @Override
            public VehiceBean map(String jsonStr) throws Exception {
                return JSON.parseObject(jsonStr, VehiceBean.class);
            }
        });

        // 定义水位线
        SingleOutputStreamOperator<VehiceBean> vehiceDS = mapDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        // 乱序
                        .<VehiceBean>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                        // 指定事件时间
                        .withTimestampAssigner(new SerializableTimestampAssigner<VehiceBean>() {
                            @Override
                            public long extractTimestamp(VehiceBean vehiceBean, long recordTimestamp) {
                                // 设置事件时间
                                return vehiceBean.getActionTime() * 1000;
                            }
                        })
        );

        // 1、按照区域进行分组
        KeyedStream<VehiceBean, String> keyByDS = vehiceDS.keyBy(value -> value.getAreaId());

        // 2、开窗     时间、窗口
        /*
            window : 必须要分组、多并行度  效率高
            wsndowAll : 可以不分组，使现实
         */

        // 定义侧输出流
        OutputTag<VehiceBean> vehiceBeanOutputTag = new OutputTag<>("vehcile_out", Types.POJO(VehiceBean.class));

        // 开窗
        SingleOutputStreamOperator<AreaBean> processDS = keyByDS
                // 滚动窗口
                .window(TumblingEventTimeWindows.of(Time.minutes(1)))
                // 关窗之后，数据到达
                .sideOutputLateData(vehiceBeanOutputTag)
                // 推迟窗口关闭时间
                .allowedLateness(Time.seconds(10))
                .process(new ProcessWindowFunction<VehiceBean, AreaBean, String, TimeWindow>() {
                    @Override
                    public void process(String key,
                                        ProcessWindowFunction<VehiceBean, AreaBean, String, TimeWindow>.Context context,
                                        Iterable<VehiceBean> elements,
                                        Collector<AreaBean> out) throws Exception {

                        // 获取上下中时间窗口
                        String startTime = DateFormatUtils.format(context.window().getStart(), "yyyy-MM-dd HH:mm:ss");
                        String endTime = DateFormatUtils.format(context.window().getEnd(), "yyyy-MM-dd HH:mm:ss");

                        // 城阳区  车数量
                        // 声明set集合  --> redis set
                        Set VehiceSet = new HashSet();

                        // 迭代窗口中所有数据
                        for (VehiceBean element : elements) {
                            VehiceSet.add(element.getCar());
                        }

                        out.collect(AreaBean.builder()
                                .areaId(key)
                                .startTime(startTime)
                                .endTime(endTime)
                                .num(VehiceSet.size())
                                .build());
                    }
                });


        // 4、退维

        // 5、写出数据HBase
        SingleOutputStreamOperator<JSONObject> areaDS = processDS.map(new MapFunction<AreaBean, JSONObject>() {
            @Override
            public JSONObject map(AreaBean areaBean) throws Exception {
                // AreaBean(areaId=120115, startTime=2025-07-29 16:36:00, endTime=2025-07-29 16:37:00, num=2)
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("op", "r");
                jsonObject.put("sinkTable", "dws_area");
                jsonObject.put("sinkFamily", "info");
                jsonObject.put("sinkRowKey", "areaId,startTime");
                jsonObject.put("after", JSON.toJSONString(areaBean));

                return jsonObject;
            }
        });

        areaDS.addSink(new HBaseSink("intct83", 1));

        // 6 启动作业
        env.execute();
    }
}
